package app.dwd;


import app.BaseSQLApp;
import common.Constant;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import utils.SQLUtil;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2022/10/13 09:24
 */
public class Dwd_05_DwdTradePayDetailSuc extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_05_DwdTradePayDetailSuc().init(
            3005,
            2,
            "Dwd_05_DwdTradePayDetailSuc"
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_05_DwdTradePayDetailSuc");

        // 2. 从 ods_db 过滤出 支付成功表数据
        Table paymentInfo = tEnv.sqlQuery("select " +
                                                "data['order_id'] order_id, " +
                                                "data['payment_type'] payment_type, " +
                                                "data['callback_time'] callback_time, " +
                                                " ts " +
                                                "from ods_db " +
                                                "where `database`='edu' " +
                                                "and `table`='payment_info' " +
                                                "and `type`='update' " +
                                                "and `old`['payment_status'] is not null " +
                                                "and `data`['payment_status']='1602'");
        tEnv.createTemporaryView("payment_info", paymentInfo);
        // 3. 从 ods_db 过滤出 订单详情表数据
        Table orderDetail = tEnv.sqlQuery("select " +
                "data['order_id'] order_id, " +
                "data['user_id'] user_id, " +
                "data['id'] id, " +
                " ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_detail' ");
        tEnv.createTemporaryView("order_detail", orderDetail);


        // 5. 2 张表进行 join
        Table result = tEnv.sqlQuery("select " +
                                         "od.id order_detail_id, " +
                                         "od.order_id, " +
                                         "od.user_id, " +
                                         "pi.payment_type payment_type_code, " +
                                         "pi.callback_time, " +
                                         "pi.ts " +
                                         "from order_detail od " +
                                         "join payment_info pi on od.order_id=pi.order_id ");
        
        // 5. 写出到 Kafka
        tEnv.executeSql("create table dwd_trade_pay_detail_suc( " +
                            "order_detail_id string, " +
                            "order_id string, " +
                            "user_id string, " +
                            "payment_type_code string, " +
                            "callback_time string, " +
                            "ts bigint " +
                            ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_PAY_DETAIL_SUC));
    
        result.executeInsert("dwd_trade_pay_detail_suc");
    }
}
